package com.liuqi.knet.remoting.transport.dispatcher.handler;

import com.liuqi.knet.remoting.Channel;
import com.liuqi.knet.remoting.ChannelHandler;
import com.liuqi.knet.remoting.RemotingException;
import com.liuqi.knet.remoting.ServerConfig;
import com.liuqi.knet.remoting.transport.AbstractChannelHandlerDelegate;
import com.liuqi.loggers.api.Logger;
import com.liuqi.loggers.facade.LoggersFacade;
import java.util.Enumeration;
import java.util.concurrent.ConcurrentHashMap;

import static com.liuqi.knet.Constants.HEARTBEAT_ATTR_KEY;

/**
 * 服务端, 对客户端连接进行校验
 *
 * 1.定时检查客户端是否发起心跳
 *
 * @author liuqi
 * @date 2025/2/8 18:10
 **/
public class ServerVerifyHandler extends AbstractChannelHandlerDelegate {

    private static final Logger logger = LoggersFacade.getLogger(ServerVerifyHandler.class);

    /**
     * 缓存所有客户端连接
     */
    private final ConcurrentHashMap<Channel, Object> channels = new ConcurrentHashMap<Channel, Object>();

    private final ServerConfig serverConfig;

    public ServerVerifyHandler(ChannelHandler handler, ServerConfig serverConfig) {
        super(handler);
        this.serverConfig = serverConfig;

        if (this.serverConfig.isEnableHeartbeat()) {
            logger.info("是否对客户端定时检测心跳[{}], 心跳间隔[{}]", true, this.serverConfig.getHeartbeatTimeout());

            new VerifyThread().start();
        }
    }

    @Override
    public void connected(Channel channel) throws RemotingException {
        super.connected(channel);

        channels.put(channel, channel);
    }

    @Override
    public void disconnected(Channel channel) throws RemotingException {
        super.disconnected(channel);

        channels.remove(channel);
    }

    @Override
    public void caught(Channel channel, Throwable exception) throws RemotingException {
        super.caught(channel, exception);

        channels.remove(channel);
    }

    /**
     * 校验心跳
     *
     * @param currentTimeMillis
     * @param ch
     */
    private void checkHandshakeHeartbeat(long currentTimeMillis, Channel ch) {
        try {
            // 检查客户端心跳是否超时
            if (currentTimeMillis - getHeartbeatTimestamp(ch) > this.serverConfig.getHeartbeatTimeout()) {
                logger.error("客户端[{}]心跳检测超时, 心跳检测间隔[{}], 主动断开客户端连接.", ch.getRemoteAddress(),
                        this.serverConfig.getHeartbeatTimeout());
                try {
                    ch.close();
                } catch (Throwable t) {
                    logger.warn(t.getMessage());
                }
            }
        } catch (Throwable t) {
            logger.error("校验客户端心跳失败. ", t);
            try {
                ch.close();
            } catch (Throwable t1) {
                logger.warn(t1.getMessage());
            }
        }
    }

    /**
     * 获取上一次心跳间隔时间
     *
     * @param ch
     * @return
     */
    public long getHeartbeatTimestamp(Channel ch) {
        Object value = ch.getAttribute(HEARTBEAT_ATTR_KEY);
        if (value == null) {
            value = getConnectTimestamp(ch);
        }
        return (Long) value;
    }

    /**
     * 获取建立连接时间
     *
     * @param ch
     * @return
     */
    public long getConnectTimestamp(Channel ch) {
        Object value = ch.getAttribute(Channel.connectTimestampKey);
        if (value == null) {
            throw new NullPointerException("未设置建立连接的时间戳. Channel: " + ch);
        }
        return (Long) value;
    }

    private class VerifyThread extends Thread {
        public VerifyThread() {
            super(new VerifyClientRunnable());
            setDaemon(true);
            setName("knet-server-verify");
        }
    }

    private class VerifyClientRunnable implements Runnable {
        @Override
        public void run() {
            while (true) {
                Enumeration<Channel> keys = channels.keys();
                while (keys.hasMoreElements()) {
                    Channel ch = keys.nextElement();
                    long current = System.currentTimeMillis();

                    // 检查是否在规定时间内, 收到心跳
                    checkHandshakeHeartbeat(current, ch);

                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                    }
                }
            }
        }
    }

}
